package com.tony.producer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

/**
 * Author tony
 * Description //TODO
 * Date 2020/8/16 17:52
 */
public class MyConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        //kafka消费的的地址
        props.put("bootstrap.servers", "121.37.25.233:9092,121.37.25.233:9092,121.37.25.233:9092");
        //组名 不同组名可以重复消费
        props.put("group.id", "GROUPID");
        //是否自动提交
        props.put("enable.auto.commit", "true");
        //从poll(拉)的回话处理时长
        props.put("auto.commit.interval.ms", "1000");
        //超时时间
        props.put("session.timeout.ms", "30000");
        //一次最大拉取的条数
        props.put("max.poll.records", 1000);
//		earliest当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
//		latest
//		当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
//		none
//		topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
        props.put("auto.offset.reset", "earliest");
        //序列化
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props );
        consumer.subscribe(Arrays.asList("bigdata"));
        ConsumerRecords<String, String> con = consumer.poll(100);
        for (ConsumerRecord<String, String> record : con) {
            System.out.println(record.key()+"==="+record.value());
        }
        consumer.close();
    }
}
